package cn.spdb.harrier.server.worker;

import java.lang.reflect.InvocationTargetException;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import javax.annotation.PostConstruct;

import org.apache.commons.lang3.ObjectUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import cn.spdb.harrier.common.Constants;
import cn.spdb.harrier.common.utils.Host;
import cn.spdb.harrier.common.utils.IPUtils;
import cn.spdb.harrier.common.utils.NameThreadFactory;
import cn.spdb.harrier.common.utils.OSUtils;
import cn.spdb.harrier.common.utils.PropertyUtils;
import cn.spdb.harrier.common.utils.SnowFlakeBuidID;
import cn.spdb.harrier.common.utils.Stopper;
import cn.spdb.harrier.common.utils.Symbol;
import cn.spdb.harrier.common.utils.ThreadUtils;
import cn.spdb.harrier.common.utils.URI;
import cn.spdb.harrier.dao.entity.UdsJobStepRecord;
import cn.spdb.harrier.dao.entity.UdsServer;
import cn.spdb.harrier.rpc.client.RpcClient;
import cn.spdb.harrier.server.entity.JobExecutionContext;
import cn.spdb.harrier.server.entity.WorkingInfo;
import cn.spdb.harrier.server.master.rpc.transport.MasterTransportServerInterfasce;
import cn.spdb.harrier.server.worker.cache.TaskExecutionContextCacheManager;
import cn.spdb.harrier.server.worker.cache.TaskExecutionContextCacheManagerImpl;
import cn.spdb.harrier.server.worker.conf.WorkerConfig;
import cn.spdb.harrier.server.worker.exec.ExecuteRunable;
import cn.spdb.harrier.server.worker.exec.TaskExecuteRunnable;
import cn.spdb.harrier.service.db.DbRegistryService;

@Component
public class WorkerManagerThread implements Runnable {

	private final DelayQueue<ExecuteRunable> workerExecuteQueue = new DelayQueue<ExecuteRunable>();

	private final Logger logger = LoggerFactory.getLogger(WorkerManagerThread.class.getSimpleName());

	private final ExecutorService workerExecService;

	private final WorkerConfig workerConfig;

	private SnowFlakeBuidID buildID;

	private WorkingInfo workingInfo = new WorkingInfo();

	private UdsServer udsServer;

	private TaskExecutionContextCacheManager taskCache = new TaskExecutionContextCacheManagerImpl();

	private ScheduledExecutorService scheduledService = Executors.newScheduledThreadPool(
			Runtime.getRuntime().availableProcessors(), new NameThreadFactory("Worker-Scheduled-Service"));

	@Autowired
	private DbRegistryService dbRegistryService;

	@Autowired
	public WorkerManagerThread(WorkerConfig workerConfig) {
		this.workerConfig = workerConfig;
		this.workerExecService = ThreadUtils.newDaemonFixedThreadExecutor("Worker-Execute-Thread",
				this.workerConfig.getWorkerExecThreads());
	}

	/**
	 * get queue size
	 *
	 * @return queue size
	 */
	public int getQueueSize() {
		return workerExecuteQueue.size();
	}

	/**
	 * submit task
	 *
	 * @param taskExecuteRunnable taskExecuteThread
	 * @return submit result
	 */
	public boolean offer(ExecuteRunable taskExecuteRunnable) {
		taskCache.cacheTaskExecutionContext(taskExecuteRunnable);
		return workerExecuteQueue.offer(taskExecuteRunnable);

	}

/**
 * 初始化并启动 Worker 管理线程及服务注册。
 *
 * 该方法使用 @PostConstruct 注解，表示在 Bean 初始化完成后自动执行。
 * 主要完成以下工作：
 * - 启动一个守护线程用于执行任务调度；
 * - 构建当前 Worker 的元信息（如 IP、端口、角色等）；
 * - 向 Master 注册自身信息；
 * - 设置定时任务定期上报 Worker 当前运行状态（CPU、内存、负载等）；
 */
@PostConstruct
public void start() {
    // 创建并启动后台线程，用于处理任务队列中的任务
    Thread thread = new Thread(this, "Worker-Execute-Manager-Thread");
    thread.setDaemon(true); // 设置为守护线程，随主线程退出而终止
    thread.start();

    // 构造当前 Worker 节点的元信息
    UdsServer udsServer = new UdsServer();
    udsServer.setPort(workerConfig.getListenPort()); // 设置监听端口
    udsServer.setServerRoleName(workerConfig.getRoleName()); // 设置角色名称
    udsServer.setServerRoleNameGroup(workerConfig.getRoleGroup()); // 设置角色组名
    udsServer.setNodeClusterType(Constants.THREAD_NAME_WORKER_SERVER); // 设置节点类型为 Worker
    udsServer.setIp(IPUtils.getHostIp()); // 获取本机 IP 地址
    udsServer.setServerName(workerConfig.getServerName()); // 设置服务名称

    // 向 Master 注册当前 Worker，并监听注册事件（目前未做具体处理）
    udsServer = dbRegistryService.registrydb(udsServer, event -> {
        // 可以在这里添加注册成功/失败的回调逻辑
    });

    // 记录注册成功的日志
    logger.info("Registry Service address:{}", udsServer.getIp() + Symbol.MAO_HAO + udsServer.getPort());

    // 保存注册后的 udsServer 实例供后续使用
    this.udsServer = udsServer;

    // 初始化 Worker 工作信息（主机地址、最大并发数、ID）
    workingInfo.setHost(new Host(udsServer.getIp(), udsServer.getPort(), udsServer.getServerName()));
    workingInfo.setJobNumMax(workerConfig.getWorkerExecThreads());
    workingInfo.setId(udsServer.getId());

    // 根据注册的 serverId 初始化雪花算法生成唯一 ID 的实例
    buildID = new SnowFlakeBuidID(udsServer.getId());

    // 启动定时任务，每隔 30 秒向 Master 汇报当前 Worker 的资源使用情况
    scheduledService.scheduleWithFixedDelay(() -> {
        try {
            sendWorkingInfo(); // 发送 Worker 运行时信息给 Master
        } catch (Exception e) {
            e.printStackTrace();
        }
    }, 30, 30, TimeUnit.SECONDS);
}


/**
 * 定期向 Master 节点发送当前 Worker 的运行状态信息。
 *
 * 该方法主要完成以下工作：
 * - 遍历所有注册的服务器节点；
 * - 筛选出可用的 Master 节点；
 * - 创建与 Master 的 RPC 连接；
 * - 获取当前 Worker 的 CPU 使用率、内存占用和负载信息；
 * - 将这些运行时信息通过 RPC 接口发送给 Master。
 */
private void sendWorkingInfo() {
    // 遍历所有服务节点，查找可用的 Master 节点
    for (UdsServer udsServer : dbRegistryService.getUdsServerList()) {
        // 判断是否为启用状态的 Master 节点
        if (udsServer.getNodeClusterType().equals(Constants.THREAD_NAME_MASTER_SERVER) && udsServer.getIsEnable()) {
            MasterTransportServerInterfasce serverInterfasce = null;
            try {
                // 创建与 Master 的 RPC 连接
                serverInterfasce = RpcClient.getInstance().create(MasterTransportServerInterfasce.class,
                        new URI("spdb", udsServer.getIp(), udsServer.getPort()));
            } catch (InstantiationException | IllegalAccessException | IllegalArgumentException
                    | InvocationTargetException | NoSuchMethodException | SecurityException e) {
                // 捕获异常并打印堆栈信息
                e.printStackTrace();
            }
            // 如果 RPC 连接创建成功，则发送 Worker 当前运行信息
            if (!ObjectUtils.isEmpty(serverInterfasce)) {
                workingInfo.setCpu(OSUtils.cpuUsage());         // 获取 CPU 使用率
                workingInfo.setMemory(OSUtils.memoryUsage());   // 获取内存使用率
                workingInfo.setLoadAverage(OSUtils.loadAverage()); // 获取系统平均负载
                serverInterfasce.updateWorkingInfo(workingInfo); // 发送信息给 Master
            }
        }
    }
}


	public void killTaskBeforeExecuteByInstanceId(Long taskInstanceId) {
		workerExecuteQueue.stream().filter(
				taskExecuteThread -> taskExecuteThread.getJobExecutionContext().getTaskInstanceId() == taskInstanceId)
				.forEach(workerExecuteQueue::remove);
		ExecuteRunable exec = taskCache.getByTaskInstanceId(taskInstanceId);
		if (ObjectUtils.isNotEmpty(exec)) {
			exec.kill();
		}
	}

	public void insertUdsJobRecord(JobExecutionContext jobContext) {
		UdsServer udsServer = getLeaderMaster();
		if (ObjectUtils.isEmpty(udsServer)) {
			return;
		}
		try {
			MasterTransportServerInterfasce serverInterfasce = RpcClient.getInstance().create(
					MasterTransportServerInterfasce.class, new URI("spdb", udsServer.getIp(), udsServer.getPort()));
			if (ObjectUtils.isEmpty(serverInterfasce)) {
				return;
			}
			serverInterfasce.insertUdsJobRecord(jobContext);
		} catch (InstantiationException | IllegalAccessException | IllegalArgumentException | InvocationTargetException
				| NoSuchMethodException | SecurityException e) {
			logger.error("insertUdsJobRecord", e);
		}
	}
	public void insertUdsJobStepRecord(UdsJobStepRecord record) {
		UdsServer udsServer = getLeaderMaster();
		if (ObjectUtils.isEmpty(udsServer)) {
			return;
		}
		try {
			MasterTransportServerInterfasce serverInterfasce = RpcClient.getInstance().create(
					MasterTransportServerInterfasce.class, new URI("spdb", udsServer.getIp(), udsServer.getPort()));
			if (ObjectUtils.isEmpty(serverInterfasce)) {
				return;
			}
			record.setAddress(workingInfo.getHost().getAddress());
			serverInterfasce.insertUdsJobStepRecord(record);
		} catch (InstantiationException | IllegalAccessException | IllegalArgumentException | InvocationTargetException
				| NoSuchMethodException | SecurityException e) {
			logger.error("insertUdsJobStepRecord", e);
		}
	}

/**
 * 通知 Master 当前任务已完成，并清理本地缓存。
 *
 * 该方法的主要作用是：
 * - 获取当前任务所属的 Master 节点；
 * - 通过 RPC 调用通知 Master 当前任务上下文已完成；
 * - 不论通知是否成功，最终都会从本地缓存中移除该任务上下文，防止内存泄漏；
 *
 * @param jobContext 当前已完成的任务执行上下文对象
 */
public void updatejobDone(JobExecutionContext jobContext) {
    // 获取当前集群中的主节点（Leader Master）
    UdsServer udsServer = getLeaderMaster();

    // 如果没有可用的 Master 节点，则直接返回
    if (ObjectUtils.isEmpty(udsServer)) {
        return;
    }

    try {
        // 创建与 Master 的 RPC 客户端连接
        MasterTransportServerInterfasce serverInterfasce = RpcClient.getInstance().create(
                MasterTransportServerInterfasce.class, new URI("spdb", udsServer.getIp(), udsServer.getPort()));

        // 如果创建失败，返回不做处理
        if (ObjectUtils.isEmpty(serverInterfasce)) {
            return;
        }

        // 通过 RPC 通知 Master 当前任务已完成
        serverInterfasce.jobContextDone(jobContext);

    } catch (InstantiationException | IllegalAccessException | IllegalArgumentException
           | InvocationTargetException | NoSuchMethodException | SecurityException e) {
        // 捕获并记录异常日志
        logger.error("更新任务完成状态时发生异常", e);
    } finally {
        // 无论是否通知成功，都清理本地缓存中的任务上下文信息
        taskCache.removeByTaskInstanceId(jobContext.getTaskInstanceId());
    }
}


	public void updateJobStepRecord(UdsJobStepRecord record) {
		UdsServer udsServer = getLeaderMaster();
		if (ObjectUtils.isEmpty(udsServer)) {
			return;
		}
		try {
			MasterTransportServerInterfasce serverInterfasce = RpcClient.getInstance().create(
					MasterTransportServerInterfasce.class, new URI("spdb", udsServer.getIp(), udsServer.getPort()));
			if (ObjectUtils.isEmpty(serverInterfasce)) {
				return;
			}
			serverInterfasce.updateJobStepRecord(record);
		} catch (InstantiationException | IllegalAccessException | IllegalArgumentException | InvocationTargetException
				| NoSuchMethodException | SecurityException e) {
			logger.error("updateJobStepRecord", e);
		}
	}

	@Override
	public void run() {
		while (Stopper.isRunning()) {
			try {
				ExecuteRunable taskExecuteRunnable = workerExecuteQueue.take();
				workerExecService.submit(taskExecuteRunnable);
			} catch (Exception e) {
				logger.error("An unexpected interrupt is happened, "
						+ "the exception will be ignored and this thread will continue to run", e);
			}
		}
	}

	public void incrementJob(JobExecutionContext jobExecutionContext) {
		workingInfo.incrementJob(jobExecutionContext);
	}

	public void decrementJob(JobExecutionContext jobExecutionContext) {
		workingInfo.decrementJob(jobExecutionContext);
	}

	public WorkingInfo getWorkingInfo() {
		return workingInfo;
	}

	public void setWorkingInfo(WorkingInfo workingInfo) {
		this.workingInfo = workingInfo;
	}

	public UdsServer getLeaderMaster() {
		return dbRegistryService.getMaster();
	}

	public long getSingleId() {
		if (ObjectUtils.isEmpty(buildID)) {
			buildID = new SnowFlakeBuidID();
		}
		return buildID.getNextId();
	}

	public WorkerConfig getWorkerConfig() {
		return workerConfig;
	}

	
}
